package com.atguigu.app.dwd;

import com.atguigu.utils.MyKafkaUtil;
import com.atguigu.utils.MysqlUtil;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @className: DwdTradeOrderRefund
 * @author: LinCong
 * @description:
 * @date: 2023/2/3 23:13
 * @version: 1.0
 */

//业务服务器（mysql）-> maxwell -> kafka -> flink(DwdTradeOrderRefund) -> kafka
public class DwdTradeOrderRefund {
    public static void main(String[] args) throws Exception {

        // TODO 1. 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        ////        1.1、开启checkpoint
//        env.enableCheckpointing(5 * 60000L, CheckpointingMode.EXACTLY_ONCE);
//        //设置checkpoint的超时时间,如果 Checkpoint在 10分钟内尚未完成说明该次Checkpoint失败,则丢弃。(默认10分钟)
//        env.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(120000L);
//        //固定延迟重启   （最多重启次数，每次重启的时间间隔）
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5000L));
////        1.2、设置状态后端
//        env.setStateBackend(new HashMapStateBackend());
//        System.setProperty("HADOOP_USER_NAME", "kevin");
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop3cluster/211126/ck");
//        1.3、设置状态TTL 生产环境设置为最大乱序程度
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));

        // TODO 3. 从 Kafka 读取 topic_db 数据，封装为 Flink SQL 表
        tableEnv.executeSql("create table topic_db(" +
                "`database` string,  " +
                "`table` string,  " +
                "`type` string,  " +
                "`data` map<string, string>,  " +
                "`old` map<string, string>,  " +
                "`proc_time` as PROCTIME(),  " +
                "`ts` string  " +
                ")" + MyKafkaUtil.getKafkaDDL("topic_db", "dwd_trade_order_refund"));

        // TODO 4. 读取退单表数据
        Table orderRefundInfo = tableEnv.sqlQuery("select  " +
                "data['id'] id,  " +
                "data['user_id'] user_id,  " +
                "data['order_id'] order_id,  " +
                "data['sku_id'] sku_id,  " +
                "data['refund_type'] refund_type,  " +
                "data['refund_num'] refund_num,  " +
                "data['refund_amount'] refund_amount,  " +
                "data['refund_reason_type'] refund_reason_type,  " +
                "data['refund_reason_txt'] refund_reason_txt,  " +
                "data['create_time'] create_time,  " +
                "proc_time,  " +
                "ts  " +
                "from topic_db  " +
                "where `table` = 'order_refund_info'  " +
                "and `type` = 'insert'  ");
        tableEnv.createTemporaryView("order_refund_info", orderRefundInfo);

        // TODO 5. 读取订单表数据，筛选退单数据
        Table orderInfoRefund = tableEnv.sqlQuery("select  " +
                "data['id'] id,  " +
                "data['province_id'] province_id,  " +
                "`old`  " +
                "from topic_db  " +
                "where `table` = 'order_info'  " +
                "and `type` = 'update'  " +
//                1005 退款中
                "and data['order_status']='1005'  " +
                "and `old`['order_status'] is not null");

        tableEnv.createTemporaryView("order_info_refund", orderInfoRefund);

        // TODO 6. 建立 MySQL-LookUp 字典表
        tableEnv.executeSql(MysqlUtil.getBaseDicLookUpDDL());

        // TODO 7. 关联三张表获得退单宽表
        Table resultTable = tableEnv.sqlQuery("select   " +
                "ri.id,  " +
                "ri.user_id,  " +
                "ri.order_id,  " +
                "ri.sku_id,  " +
                "oi.province_id,  " +
                "date_format(ri.create_time,'yyyy-MM-dd') date_id,  " +
                "ri.create_time,  " +
                "ri.refund_type,  " +
                "type_dic.dic_name,  " +
                "ri.refund_reason_type,  " +
                "reason_dic.dic_name,  " +
                "ri.refund_reason_txt,  " +
                "ri.refund_num,  " +
                "ri.refund_amount,  " +
                "ri.ts,  " +
                "current_row_timestamp() row_op_ts  " +
                "from order_refund_info ri  " +
                "join   " +
                "order_info_refund oi  " +
                "on ri.order_id = oi.id  " +
                "join   " +
                "base_dic for system_time as of ri.proc_time as type_dic  " +
                "on ri.refund_type = type_dic.dic_code  " +
                "join  " +
                "base_dic for system_time as of ri.proc_time as reason_dic  " +
                "on ri.refund_reason_type=reason_dic.dic_code");
        tableEnv.createTemporaryView("result_table", resultTable);

        // TODO 8. 建立 Kafka-Connector dwd_trade_order_refund 表
        tableEnv.executeSql("create table dwd_trade_order_refund(  " +
                "id string,  " +
                "user_id string,  " +
                "order_id string,  " +
                "sku_id string,  " +
                "province_id string,  " +
                "date_id string,  " +
                "create_time string,  " +
                "refund_type_code string,  " +
                "refund_type_name string,  " +
                "refund_reason_type_code string,  " +
                "refund_reason_type_name string,  " +
                "refund_reason_txt string,  " +
                "refund_num string,  " +
                "refund_amount string,  " +
                "ts string,  " +
                "row_op_ts timestamp_ltz(3)  " +
                ")" + MyKafkaUtil.getKafkaSinkDDL("dwd_trade_order_refund"));

        // TODO 9. 将关联结果写入 Kafka-Connector 表
        tableEnv.executeSql("" +
                "insert into dwd_trade_order_refund select * from result_table");
    }
}
